/*
   Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
   Original sources are available at www.latestbit.com

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package org.latestbit.smartobjects.network.tcp;

import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Collection;
import java.util.Date;
import java.util.LinkedList;
import java.util.concurrent.locks.ReentrantLock;

import org.latestbit.smartobjects.encrypters.IDataEncrypter;
import org.latestbit.smartobjects.exceptions.HandshakeException;
import org.latestbit.smartobjects.exceptions.InvalidPacketReceivedException;
import org.latestbit.smartobjects.io.ResizableByteBuffer;
import org.latestbit.smartobjects.network.IConnection;
import org.latestbit.smartobjects.network.IUserCredentials;
import org.latestbit.smartobjects.network.IConnectionListener;
import org.latestbit.smartobjects.network.IHandshakeManager;
import org.latestbit.smartobjects.network.IPacketFormer;
import org.latestbit.smartobjects.network.IStreamListener;
import org.latestbit.smartobjects.utils.IThreadPool;
import org.latestbit.smartobjects.utils.PostActionListenersContainer;

public abstract class TCPConnection implements IConnection, ISocketMultiplexerListener, Runnable {
	
	private SocketChannel channel;
	protected ReentrantLock channelLock = new ReentrantLock();
	
	protected Collection<IConnectionListener> connectionListeners = new LinkedList<IConnectionListener>();
	protected PostActionListenersContainer<IConnectionListener> palConnectionListeners = new PostActionListenersContainer<IConnectionListener> ();
	
	protected Collection<IStreamListener> streamListeners = new LinkedList<IStreamListener>();
	protected PostActionListenersContainer<IStreamListener> palStreamListeners = new PostActionListenersContainer<IStreamListener> ();
	
	private URI endpointURI;
	protected TCPSharedResources resources;
	private SocketMultiplexer multiplexer;
	private final String CONNECTION_MULTIPLEXER_GROUP_NAME = "CONNECTIONS";
	private SelectionKey channelKey;
	
	private final int KEEP_ALIVE_MAX_INACTIVITY_TIMEOUT = 3000; // milliseconds
	private Date lastActivityTime = new Date();
	
	private final int READ_BUFFER_SIZE = 65535;
	// Direct buffer is working too bad on some platforms (Android, Solaris?)
	// private ByteBuffer readBuffer = ByteBuffer.allocateDirect(READ_BUFFER_SIZE);
	private ByteBuffer readBuffer = ByteBuffer.allocate(READ_BUFFER_SIZE);
	
	private final String CONNECTION_THREAD_POOL_GROUP_NAME = "CONNECTION_WRITERS";	
	private IThreadPool writerThreadPool;
	private final int WRITE_BUFFER_SIZE = 65535;
	// Direct buffer is working too bad on some platforms (Android, Solaris?)
	// private ResizableByteBuffer writeBuffer =  new ResizableByteBuffer (true, WRITE_BUFFER_SIZE, Integer.MAX_VALUE, WRITE_BUFFER_SIZE );
	private ResizableByteBuffer writeBuffer =  new ResizableByteBuffer (false, WRITE_BUFFER_SIZE, Integer.MAX_VALUE, WRITE_BUFFER_SIZE ); 
	
	protected IPacketFormer packetFormer;	
	protected IHandshakeManager handshakeManager;
	private IDataEncrypter dataEncrypter;
	private IUserCredentials userCrentials;
	
	
	public TCPConnection(TCPSharedResources resources) throws IOException {
		this.resources = resources;
		this.resources.addRef();
		this.multiplexer = resources.getSocketMultiplexer(CONNECTION_MULTIPLEXER_GROUP_NAME);
		this.writerThreadPool = resources.getThreadPool(CONNECTION_THREAD_POOL_GROUP_NAME);

		if(resources.getPacketFormerFactory()!=null)
			this.packetFormer = resources.getPacketFormerFactory().createPacketFormer();
		if(resources.getHandshakeManagerFactory()!=null)
			this.handshakeManager = resources.getHandshakeManagerFactory().createHandshakeManager(this);		
	}
	
	public void finalize() {
		destroy();
	}
	
	@Override
	public void destroy() {
		close();
		this.resources.release();
	}

	@Override
	public void addConnectionListener(IConnectionListener listener) {
		synchronized(connectionListeners) {
			if(!palConnectionListeners.isActivated()) {
				connectionListeners.add(listener);
			}
			else
				palConnectionListeners.putListenerToAdd(listener);
		}
	}

	@Override
	public void delConnectionListener(IConnectionListener listener) {
		synchronized(connectionListeners) {
			if(!palConnectionListeners.isActivated()) {
				connectionListeners.remove(listener);
			}
			else
				palConnectionListeners.putListenerToRemove(listener);				
		}
	}
	
	@Override
	public void addStreamListener(IStreamListener listener) {
		synchronized(streamListeners) {
			if(!palStreamListeners.isActivated()) {
				streamListeners.add(listener);
			}
			else
				palStreamListeners.putListenerToAdd(listener);
		}
	}

	@Override
	public void delStreamListener(IStreamListener listener) {
		synchronized(streamListeners) {
			if(!palStreamListeners.isActivated()) {
				streamListeners.remove(listener);
			}
			else
				palStreamListeners.putListenerToRemove(listener);				
		}
	}
	

	@Override
	public URI getEndpointURI() {
		return endpointURI;
	}
	
	public void setEndpointURI(URI value) {
		this.endpointURI = value;
	}


	public void setChannel(SocketChannel channel) {
		this.channel = channel;
	}

	public SocketChannel getChannel() {
		return channel;
	}

	@Override
	public boolean isOpen() {
		boolean result = false;
		channelLock.lock();
		try {
			result = channel != null && channel.isOpen();			
		}
		finally {
			channelLock.unlock();
		}
		return result;
	}
	
	@Override
	public void close() {
		channelLock.lock();
		try {
	        if(isOpen()) {
	        	
	        	try {
					channel.socket().shutdownInput();
					channel.socket().shutdownOutput();
				} 
	        	catch (IOException e) {};
	        	
	            try {
	            	channel.close();
	            }
	            catch (IOException e) {}            
	        }
	        if(channelKey!=null && channelKey.isValid()) {
	        	multiplexer.removeChannel(channelKey);
	        	channelKey = null;
	        }
	        if(this.handshakeManager!=null) {
	        	this.handshakeManager.reset();
	        }
	        if(this.packetFormer!=null) {
	        	this.packetFormer.reset();
	        }	        
		}
		finally {
			channelLock.unlock();
		}
    }

	public void start() throws IOException {
		channelLock.lock();
		try {
			this.channel.configureBlocking(false);
			lastActivityTime = new Date();
			channelKey = multiplexer.addChannel(channel, SelectionKey.OP_READ, this);
		}
		finally {
			channelLock.unlock();
		}
	}

	@Override
	public void onSelectedEvent(SelectionKey selectionKey) {
		int len = 0;		
		if(isOpen()) {
			do {
				try {
					channelLock.lock();
					lastActivityTime = new Date();
					try {
						if(channel!=null)
							len = this.channel.read(readBuffer);
					}
					finally {
						channelLock.unlock();
					}
				}
				catch (IOException e) {
					fireOnReceivedError(e.toString());
				}
				if(len > 0) {
					readBuffer.flip();
					
					if(this.packetFormer==null) {						
						fireOnReceived(readBuffer);						
					}
					else {
						this.packetFormer.onReceiveRawPacket(readBuffer);
						
						ByteBuffer formedBuffer = null;
						try {
							do {
								formedBuffer = this.packetFormer.getReadyPacket();
																
								
								if(formedBuffer!=null) {									
									fireOnReceived(formedBuffer);
								};
							}
							while(formedBuffer!=null);
						}
						catch(InvalidPacketReceivedException ex) {
							fireOnReceivedError(ex.toString());
							fireOnDisconnected();
							close();							
						}
					}
					readBuffer.clear();
				}
				else {
					if(len < 0 ) {
						fireOnReceivedError("Connection forcefully closed");
					}
					fireOnDisconnected();
					close();
				};
			}
			while(len > READ_BUFFER_SIZE );
		}
	}
	
	public void onCheckKeepAlive(SelectionKey forKey) {
		Date now = new Date();
		channelLock.lock();
		try {
			if(isOpen()) {
				try {
					if(lastActivityTime.getTime() + KEEP_ALIVE_MAX_INACTIVITY_TIMEOUT < now.getTime() ) {
						lastActivityTime = now;
						channel.socket().sendUrgentData(0xAA);						
					}
				} 
				catch (IOException e) {}
			}			
		}
		finally {
			channelLock.unlock();
		}
	}
	
	protected abstract void fireOnConnected();

	protected void fireOnDisconnected() {
		if(this.handshakeManager==null || this.handshakeManager.isHandshaked()) {
			synchronized (connectionListeners) {
				palConnectionListeners.activate();
				try {				
					for(IConnectionListener listener : connectionListeners) {
						listener.onDisconnected(this);
					}
				}
				finally {
					processingPostActionConnectionListeners();
				}
			}
		}
	}

	protected void fireOnReceivedError(String errorMessage) {
		if(this.handshakeManager==null || ( this.handshakeManager!=null && !this.handshakeManager.isHandshaked())) {
			synchronized (streamListeners) {
				palStreamListeners.activate();
				try {
					for(IStreamListener listener : streamListeners) {
						listener.onReceiveError(this, errorMessage);
					}
				}
				finally {
					processingPostActionStreamListeners();
				}
			}
		}
	}
	
	private void fireOnWriteError(String errorMessage) {
		if(this.handshakeManager==null || ( this.handshakeManager!=null && !this.handshakeManager.isHandshaked())) {
			synchronized (streamListeners) {
				palStreamListeners.activate();
				try {
					for(IStreamListener listener : streamListeners) {
						listener.onWriteError(this, errorMessage);
					}
				}
				finally {
					processingPostActionStreamListeners();
				}				
			}
		}
	}	

	protected void fireOnReceived(ByteBuffer buffer) {
		if(this.handshakeManager!=null && !this.handshakeManager.isHandshaked()) {
			try {
				this.handshakeManager.onHandshakeReceived ( buffer );
				if(this.handshakeManager.isHandshaked()) {
					this.fireOnConnected();
				}
			}
			catch(HandshakeException ex) {
				close();
			}
		}
		else {
			synchronized (streamListeners) {
				palStreamListeners.activate();
				try {
					int curPos = buffer.position();
					int curLimit = buffer.limit();
					
					for(IStreamListener listener : streamListeners) {
						listener.onReceived(this, buffer);
						buffer.position(curPos);
						buffer.limit(curLimit);
					}
				}
				finally {
					processingPostActionStreamListeners();
				}
			}
		}
	}
	
	protected void processingPostActionConnectionListeners() {
		palConnectionListeners.deactivate();
		try {
			for ( PostActionListenersContainer<IConnectionListener>.PostActionQueueItem item : palConnectionListeners.getPostActionQueue() ) {
				if( item.isActionAdd() ) {
					addConnectionListener ( item.getListener() );
				}
				else
				if( item.isActionRemove() ) {
					delConnectionListener ( item.getListener() );
				}
			}
		}
		finally {
			palConnectionListeners.reset();
		}	
	}
	
	protected void processingPostActionStreamListeners() {
		palStreamListeners.deactivate();
		try {
			for ( PostActionListenersContainer<IStreamListener>.PostActionQueueItem item : palStreamListeners.getPostActionQueue() ) {
				if( item.isActionAdd() ) {
					addStreamListener ( item.getListener() );
				}
				else
				if( item.isActionRemove() ) {
					delStreamListener ( item.getListener() );
				}
			}
		}
		finally {
			palStreamListeners.reset();
		}	
	}
	
	@Override
	public void send(ByteBuffer buffer) throws IOException {
		
		boolean isNeededScheduling = true;

		
		
		synchronized(writeBuffer) {
			if(!isOpen())
				throw new IOException("Unable to send packet for closed connection!");
						
			isNeededScheduling = writeBuffer.getBuffer().position() == 0;

			
			writeBuffer.put ( buffer );
		}
		
				
		if ( isNeededScheduling ) {
			writerThreadPool.execute ( this );
		}		
	}

	@Override
	public void run() {
		try {
			synchronized(writeBuffer) {
				try {				
					writeBuffer.getBuffer().flip();
					boolean trying = false;
					do {				
						int bytesSent = 0;
						trying = false;
						channelLock.lock();
						try {
							if(channel!=null)
								bytesSent = channel.write(writeBuffer.getBuffer());
							else
								throw new IOException("Unable to write to closed channel");
						}
						finally {
							channelLock.unlock();
						}
						
						if(bytesSent < 0 ) {
							fireOnWriteError("Unable to write to socket");
						}
						else
							if(writeBuffer.getBuffer().position()!=writeBuffer.getBuffer().limit()) {
								trying = true;
								Thread.yield();
							}
					}
					while(trying);
				}			
				finally {
					writeBuffer.getBuffer().clear();
				}				
			}
		}
		catch (IOException e) {
			fireOnWriteError(e.toString());
		}
	}	

	@Override
	public IDataEncrypter getDataEncrypter() {
		return this.dataEncrypter;
	}

	@Override
	public void setDataEncrypter(IDataEncrypter encrypter) {
		this.dataEncrypter = encrypter;
	}

	@Override
	public IUserCredentials getUserCredentials() {
		return this.userCrentials;
	}
	
	@Override
	public void setUserCredentials(IUserCredentials credentials) {
		this.userCrentials = credentials;
	}

	@Override
	public IHandshakeManager getHandshakeManager() {
		return this.handshakeManager;
	}
	
	@Override
	public IPacketFormer getPacketFormer() {
		return this.packetFormer;
	}
	
}
